// the basic algorithms goes as we read the records in the bufs and then 
// and sort them in an external array list , then create heap file to hold the new data
// and then to open those multi pages on the bufs and get next till a page is empty
// if the page is empty we will read the next page from the iterator(getNext) method and put it in 
// the corresponding buf

// compare type = 1 ascending, compare type =0 decending
package iterator;

import global.AttrType;
import global.Convert;
import global.PageId;
import global.RID;
import global.SystemDefs;
import global.TupleOrder;
import heap.FieldNumberOutOfBoundException;
import heap.HFBufMgrException;
import heap.HFDiskMgrException;
import heap.HFException;
import heap.HFPage;
import heap.Heapfile;
import heap.InvalidSlotNumberException;
import heap.InvalidTupleSizeException;
import heap.InvalidTypeException;
import heap.Scan;
import heap.SpaceNotAvailableException;
import heap.Tuple;

import java.io.File;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Scanner;

import bufmgr.PageNotReadException;

import diskmgr.Page;

public class ExternalSort extends Sort {

	private PageId[] PageIds;
	private byte[][] bufs;// Temporarily holder for the pages that i'll use in
							// the buffer
	private TupleOrder order;
	private ArrayList<String> names = new ArrayList<String>(100);
	private ArrayList<Object> listToSort = new ArrayList<Object>(100);
	private int type = 0; // zero for string else for integer
	private Heapfile HFFinal;

	public ExternalSort(String Path, TupleOrder tu, int n_pages, AttrType[] in)
			throws Exception {
		Heapfile hff = createHeapFile("ana.txt");
		ExternalIterator it = new ExternalIterator(hff);
		order = tu;
		bufs = new byte[n_pages][1024];
		PageIds = new PageId[n_pages];
		try {
			get_buffer_pages(n_pages, PageIds, bufs);
		} catch (IteratorBMException e) {
			e.printStackTrace();
			System.err.println("failure to get the pages from the iterator");
		}

		// the first page i'll use to read from file.

		if (in[0].attrType == AttrType.attrInteger) {
			type = 1;
		} else {
			type = 0;
		}

		RID rid = new RID();// rid used to watch
		// if the HFpage is full

		int bufcnt = 0;// counts which buf i have used
		HFPage hfp = new HFPage();
		hfp.init(PageIds[bufcnt], new Page(bufs[bufcnt]));// Initializing the
															// first page
		bufcnt++;// counts which buffer i used
		Tuple temp = it.get_next();
		int namCnt = 0;// name counter used to track names
		int recordCnt = 0;// for test only
		while (rid != null && temp != null) {
			rid = hfp.insertRecord(temp.getTupleByteArray());

			if (rid == null) {// if the hfpage is full
								// if null the page is full get ready for the
								// heapfile
				Heapfile hf = new Heapfile("" + namCnt);
				names.add("" + namCnt);// adding the name to the array list
				namCnt++;
				listToSort = it.sort(listToSort, type);
				insertIntoHeapFile(listToSort, hf, type);// added the file and
				// all its
				// System.out.println("\n \n Heap File: " + namCnt);
				// print(hf);
				// records
				listToSort.clear();
				if (bufcnt == bufs.length)// if i exceed the limit, if i filled
											// all the
				// allowed space for buffering
				{
					bufcnt = 0;
				}
				// creating the new hfpage
				hfp.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
				bufcnt++;
				rid = hfp.insertRecord(temp.getTupleByteArray());

			}
			if (type == 0) {
				listToSort.add(Convert.getStrValue(0, temp.getTupleByteArray(),
						temp.getLength()));

			}
			// i didn't take input from the user as integers
			else {
				listToSort
						.add(Convert.getIntValue(0, temp.getTupleByteArray()));
			}

			temp = it.get_next();
			recordCnt++;
		}
		// new entry // write the last unwritten page

		Heapfile hf = new Heapfile("" + namCnt);
		names.add("" + namCnt);// adding the name to the array list
		namCnt++;
		listToSort = it.sort(listToSort, type);
		insertIntoHeapFile(listToSort, hf, type);// added the file and
		// all its
		// System.out.println("Heap File: " + namCnt);
		// print(hf);
		// records
		listToSort.clear();
		// end new entry

		// start the merging
		bufcnt = 0;

		// System.out.println("Inserted: " + recordCnt + " Records");

		while (names.size() > 1) {
			ArrayList<String> tempname = new ArrayList<String>();
			for (int i = 0; i + 1 < names.size(); i += 2) {
				Heapfile heNew = new Heapfile("" + namCnt);
				tempname.add("" + namCnt);
				namCnt++;

				Heapfile he1 = new Heapfile(names.get(i));
				ExternalIterator it1 = new ExternalIterator(he1);

				Heapfile he2 = new Heapfile(names.get(i + 1));
				ExternalIterator it2 = new ExternalIterator(he2);

				HFPage hfp1 = new HFPage();
				hfp1.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
				readIntoHFPage(he1, it1, hfp1);

				bufcnt++;
				HFPage hfp2 = new HFPage();
				hfp2.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
				readIntoHFPage(he2, it2, hfp2);

				PageIterator pit1 = new PageIterator(hfp1);
				PageIterator pit2 = new PageIterator(hfp2);
				Tuple temp1 = null;
				Tuple temp2 = null;
				temp1 = pit1.getNext();
				temp2 = pit2.getNext();
				int cnt = 0;
				int tem1 = 0, tem2 = 0;
				while (temp1 != null && temp2 != null) {// if temp1 < temp2 must
														// return true if
														// compare type is 0
					++cnt;

					if (compare(temp1, temp2, type, 1))// compare type ascending
														// or decending
					{
						++tem1;
						// System.out.println("Inserting :"
						// + Convert.getStrValue(0,
						// temp1.getTupleByteArray(),
						// temp1.getTupleByteArray().length));
						heNew.insertRecord(temp1.getTupleByteArray());
						temp1 = pit1.getNext();
					} else {
						++tem2;
						// System.out.println("Inserting :"
						// + Convert.getStrValue(0,
						// temp2.getTupleByteArray(),
						// temp2.getTupleByteArray().length));
						heNew.insertRecord(temp2.getTupleByteArray());
						temp2 = pit2.getNext();
					}
					if (temp1 == null) {
						bufcnt++;
						if (bufcnt == bufs.length) {
							bufcnt = 0;
						}
						hfp1.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
						if (!readIntoHFPage(he1, it1, hfp1))
							break;
						pit1 = new PageIterator(hfp1);
						temp1 = pit1.getNext();
					}
					if (temp2 == null) {
						bufcnt++;
						if (bufcnt == bufs.length) {
							bufcnt = 0;
						}

						hfp2.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
						if (!readIntoHFPage(he2, it2, hfp2))
							break;
						pit2 = new PageIterator(hfp2);
						temp2 = pit2.getNext();
					}
				}
				System.out.println("1st " + tem1 + " 2nd " + tem2);
				// new entry//add the rest of the records of page 1
				while (true) {
					if (temp1 == null) {
						bufcnt++;
						if (bufcnt == bufs.length) {
							bufcnt = 0;
						}
						hfp1.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
						if (!readIntoHFPage(he1, it1, hfp1))
							break;
						pit1 = new PageIterator(hfp1);
						temp1 = pit1.getNext();
					}
					System.out.println(++cnt);// TODO
					// System.out.println("Inserting :"
					// + Convert.getStrValue(0, temp1.getTupleByteArray(),
					// temp1.getTupleByteArray().length));
					heNew.insertRecord(temp1.getTupleByteArray());
					temp1 = pit1.getNext();
				}
				while (true) {
					if (temp2 == null) {
						bufcnt++;

						if (bufcnt == bufs.length) {
							bufcnt = 0;
						}
						hfp2.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
						if (!readIntoHFPage(he2, it2, hfp2))
							break;
						pit1 = new PageIterator(hfp1);
						temp1 = pit1.getNext();
					}
					System.out.println(++cnt);// TODO
					// System.out.println("Inserting :"
					// + Convert.getStrValue(0, temp2.getTupleByteArray(),
					// temp2.getTupleByteArray().length));
					heNew.insertRecord(temp2.getTupleByteArray());
					temp2 = pit2.getNext();
				}
				// System.out.println("THE FILE");
				// print(heNew);
				// System.out.println("END OF FILE");
				he1.deleteFile();
				he2.deleteFile();
				// end new entry
				// System.out.println(heNew.toString());
			}

			if (names.size() % 2 != 0) {
				tempname.add(names.get(names.size() - 1));
			}
			names = tempname;
		}

		HFFinal = new Heapfile(names.get(0));

		Scan scan = new Scan(HFFinal);
		RID rid2 = new RID();
		System.out.println("BEGIN");
		while (true) {
			Tuple t = scan.getNext(rid2);
			if (t == null) {
				break;
			}
			if(type==0)
			 System.out.println(Convert.getStrValue(0, t.getTupleByteArray(),
			 t.getTupleByteArray().length));
			else {
				 System.out.println(Convert.getIntValue(0, t.getTupleByteArray()));
			}
		}
	}

	private void print(Heapfile hf, int type) throws Exception {
		ExternalIterator scan = new ExternalIterator(hf);

		while (true) {
			Tuple t = scan.get_next();
			if (t == null) {
				break;
			}
			if (type == 0)
				System.out.println(Convert.getStrValue(0,
						t.getTupleByteArray(), t.getTupleByteArray().length));
			else
				System.out
						.println(Convert.getIntValue(0, t.getTupleByteArray()));

		}
	}

	public ExternalSort(AttrType[] in, short len_in, short[] str_sizes,
			Iterator am, int sort_fld, TupleOrder sort_order, int sort_fld_len,
			int n_pages) throws InvalidTupleSizeException,
			InvalidTypeException, PageNotReadException, JoinsException,
			IndexException, TupleUtilsException, PredEvalException,
			LowMemException, UnknowAttrType, UnknownKeyTypeException, Exception {

		super(in, len_in, str_sizes, am, sort_fld, sort_order, sort_fld_len,
				n_pages);
		ExternalIterator it = new ExternalIterator(createHeapFile("ana.txt"));
		bufs = new byte[n_pages][];
		try {
			get_buffer_pages(n_pages, PageIds, bufs);
		} catch (IteratorBMException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			System.err.println("failure to get the pages from the iterator");
		}

		// the first page i'll use to read from file.

		if (in[0].attrType == AttrType.attrInteger) {
			type = 1;
		} else {
			type = 0;
		}

		RID rid = new RID();// rid used to watch
		// if the HFpage is full

		int bufcnt = 0;// counts which buf i have used
		HFPage hfp = new HFPage();
		hfp.init(PageIds[bufcnt], new Page(bufs[bufcnt]));// Initializing the
															// first page
		bufcnt++;// counts which buffer i used
		Tuple temp = it.get_next();
		// Tuple temp =null;
		int namCnt = 0;// name counter used to track names
		while (rid != null && temp != null) {

			rid = hfp.insertRecord(temp.getTupleByteArray());

			if (rid == null) {// if the hfpage is full
								// if null the page is full get ready for the
								// heapfile

				Heapfile hf = new Heapfile("" + namCnt);
				names.add("" + namCnt);// adding the name to the array list
				namCnt++;
				listToSort = it.sort(listToSort, type);
				insertIntoHeapFile(listToSort, hf, type);// added the file and
															// all its
				// records
				listToSort.clear();
				if (bufcnt == bufs.length)// if i exceed the limit, if i filled
											// all the
				// allowed space for buffering
				{
					bufcnt = 0;
				}
				bufcnt++;// creating the new hfpage
				hfp.init(PageIds[bufcnt], new Page(bufs[bufcnt]));

				rid = hfp.insertRecord(temp.getTupleByteArray());

			}

			if (type == 0) {
				listToSort.add(Convert.getStrValue(0, temp.getTupleByteArray(),
						temp.getLength()));

			}
			// i didn't take input from the user as integers

			temp = it.get_next();
		}
		// start the merging
		bufcnt = 0;

		while (names.size() > 1) {
			ArrayList<String> tempname = new ArrayList<String>();
			for (int i = 0; i < names.size() / 2; i++) {
				namCnt++;
				Heapfile heNew = new Heapfile("" + namCnt);
				tempname.add("" + namCnt);
				Heapfile he1 = new Heapfile(names.get(i));
				ExternalIterator it1 = new ExternalIterator(he1);
				Heapfile he2 = new Heapfile(names.get(i + 1));
				ExternalIterator it2 = new ExternalIterator(he2);
				HFPage hfp1 = new HFPage();
				hfp1.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
				readIntoHFPage(he1, it1, hfp1);
				bufcnt++;
				HFPage hfp2 = new HFPage();
				hfp2.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
				readIntoHFPage(he2, it2, hfp2);
				PageIterator pit1 = new PageIterator(hfp1);
				PageIterator pit2 = new PageIterator(hfp2);
				Tuple temp1 = null;
				Tuple temp2 = null;
				temp1 = pit1.getNext();
				temp2 = pit2.getNext();
				while (temp1 != null && temp2 != null) {
					if (compare(temp1, temp2, type, 0)) {
						heNew.insertRecord(temp1.getTupleByteArray());
					} else {
						heNew.insertRecord(temp2.getTupleByteArray());
					}
					if ((temp1 = pit1.getNext()) == null) {
						bufcnt++;
						if (bufcnt == bufs.length) {
							bufcnt = 0;
						}
						hfp1.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
						if (!readIntoHFPage(he1, it1, hfp1))
							break;
					}
					if ((temp2 = pit2.getNext()) == null) {
						if (bufcnt == bufs.length) {
							bufcnt = 0;
						}
						bufcnt++;
						hfp2.init(PageIds[bufcnt], new Page(bufs[bufcnt]));
						if (!readIntoHFPage(he2, it2, hfp2))
							break;
					}
				}
			}
			names = tempname;
		}
	}

	// if the descending t1> t2 return true, else return false
	// if the ascending t1< t2 return true, else return false
	// private boolean compare(Tuple t1, Tuple t2, int type, int comparetype)
	// {
	// return false;
	// }

	// return if the insertion was successful
	private boolean readIntoHFPage(Heapfile hf, ExternalIterator it, HFPage hfp) {
		Tuple temp;
		try {
			temp = it.get_next();
			if (temp == null) {
				return false;
			}
			while (temp != null) {
				RID rid = hfp.insertRecord(temp.getTupleByteArray());
				if (rid == null) {
					return true;
				}
				temp = it.get_next();
			}
			return true;

		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}
	}

	// insert every record inside the array list into the heap file, type =0
	// string else integer
	private void insertIntoHeapFile(ArrayList<Object> list, Heapfile hf,
			int type) throws Exception {
		for (int i = 0; i < list.size(); i++) {
			if (type == 0) {
				String length = (String) list.get(i);
				byte array[] = new byte[length.length() + 2];
				Convert.setStrValue(length, 0, array);
				hf.insertRecord(array);
			} else {
				int x = (Integer) list.get(i);
				byte array[] = new byte[5];
				Convert.setIntValue(x, 0, array);
				hf.insertRecord(array);
			}
		}
		list.clear();// clearing the array list
	}

//	// copies from the HFPage to the Heapfile//TODO not used
//	private void Copy(HFPage hp, Heapfile hf)
//			throws InvalidSlotNumberException, IOException {
//		Tuple temp = new Tuple();
//		PageIterator it = new PageIterator(hp);
//		while ((temp = it.getNext()) != null) {
//			try {
//				hf.insertRecord(temp.getTupleByteArray());
//			} catch (Exception e) {
//				System.err.println("Error at inserting record");
//				e.printStackTrace();
//			}
//		}
//	}

	public Heapfile createHeapFile(String pathOfTheFile) throws HFException,
			HFBufMgrException, HFDiskMgrException, IOException,
			FieldNumberOutOfBoundException, InvalidSlotNumberException,
			InvalidTupleSizeException, SpaceNotAvailableException {

		// will contain the path of the file that i will read from use the
		// instance heapo to create the new heap file.
		Heapfile file = null;
		new SystemDefs("DataBase", 5000, 5000, "Clock");
		File f = new File(pathOfTheFile);
		file = new Heapfile("hi");
		Scanner scan = new Scanner(f);

		String key = scan.next();
		try {
			Integer.parseInt(key);
			InsertIntegersRecords(file, f);
		} catch (Exception e) {
			InsertStringsRecords(file, f);

		}

		// check the insertion of records
//		Scan scano = new Scan(file);
//		RID rid = new RID();
//		Tuple t = scano.getNext(rid);
//		while (t != null) {
//
//			// System.out.println(Convert.getStrValue(0, t.getTupleByteArray(),
//			// t.getTupleByteArray().length));
//
//			// System.out.println(Convert.getIntValue(0,
//			// t.getTupleByteArray()));
//
//			t = scano.getNext(rid);
//
//		}
		return file;
	}

	private void InsertIntegersRecords(Heapfile file, File f)
			throws InvalidSlotNumberException, InvalidTupleSizeException,
			SpaceNotAvailableException, HFException, HFBufMgrException,
			HFDiskMgrException, IOException {
		System.out.println("Insertion of Integers");
		Scanner scan = new Scanner(f);
		while (scan.hasNext()) {
			// Object key = scan.next();
			// System.out.println(key);
			int temp = scan.nextInt();
			byte[] array = new byte[5];
			Convert.setIntValue(temp, 0, array);
			file.insertRecord(array);
		}
	}

	// inserts string records into the heap files
	private void InsertStringsRecords(Heapfile file, File f)
			throws IOException, InvalidSlotNumberException,
			InvalidTupleSizeException, SpaceNotAvailableException, HFException,
			HFBufMgrException, HFDiskMgrException {
		System.out.println("Insertion of Strings");
		Scanner scan = new Scanner(f);
		while (scan.hasNext()) {
			Object key = scan.next();
			// System.out.println(key);
			String temp = (String) key;
			byte[] array = new byte[temp.length() + 2];
			Convert.setStrValue(temp, 0, array);
			file.insertRecord(array);
		}

	}

	class PageIterator {
		HFPage hf;
		RID cRID = new RID();
		boolean first = true;

		public PageIterator(HFPage input) throws IOException {
			hf = input;
			cRID = input.firstRecord();
		}

		public Tuple getNext() throws InvalidSlotNumberException, IOException {
			if (cRID == null)
				return null;
			Tuple rTuple = hf.getRecord(cRID);
			cRID = hf.nextRecord(cRID);
			return rTuple;
		}

		public void print(int type) throws InvalidSlotNumberException, IOException {
			Tuple t = getNext();
			while (t != null) {
				if(type==0)
				System.out.println(Convert.getStrValue(0,
						t.getTupleByteArray(), t.getTupleByteArray().length));
				else
					System.out.println(Convert.getIntValue(0,
							t.getTupleByteArray()));
				t = getNext();
			}
		}
	}

	private boolean compare(Tuple t1, Tuple t2, int type, int comparetype)
			throws IOException {
		if (type == 0)// this means the tuples are strings
		{
			String tuple1 = Convert.getStrValue(0, t1.getTupleByteArray(),
					t1.getTupleByteArray().length);
			String tuple2 = Convert.getStrValue(0, t2.getTupleByteArray(),
					t2.getTupleByteArray().length);
			System.out.println("Comparing t1: " + tuple1 + " with t2: "
					+ tuple2);
			if (comparetype == 0)// return true if t1 > t2
			{
				if (tuple1.compareTo(tuple2) > 0)
					return true;
				return false;
			} else
			// return true if t1 < t2
			{
				if (tuple1.compareTo(tuple2) < 0)
					return true;
				return false;
			}
		} else
		// this means the tuples are integers
		{
			int tuple1 = Convert.getIntValue(0, t1.getTupleByteArray());
			int tuple2 = Convert.getIntValue(0, t2.getTupleByteArray());
			System.out.println("Comparing t1: " + tuple1 + " with t2: "
					+ tuple2);
			if (comparetype == 0)// return true if t1 > t2
			{
				if (tuple1 > tuple2)
					return true;
				return false;
			} else
			// return true if t1 < t2
			{
				if (tuple1 < tuple2)
					return true;
				return false;
			}
		}
	}

	public static void main(String[] args) throws Exception {

		AttrType[] in = { new AttrType(0) };
		short[] str_sizes = { 15 };
		// ExternalIterator y = new ExternalIterator(new Heapfile("Open"));
		// short a = 1, b = 15, c = 0, d = 1, e = 10;
		TupleOrder as = new TupleOrder(TupleOrder.Ascending);
		// ExternalSort sort = new ExternalSort(in, a, str_sizes, y, 0, as,
		// 15,10);
		ExternalSort sort = new ExternalSort("ana.txt", as, 10, in);

	}
}
// John Aziz wasn't here
// Page page = new Page(bufs[0]);
// HFPage hf = new HFPage(page);
// hf.getHFpageArray();